package com.etlmaster.executor.utils;

import java.io.IOException;
import java.security.PrivilegedActionException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.security.auth.login.LoginException;

import org.apache.hadoop.conf.Configuration;

import com.etlmaster.core.DataSourceDefine;
import com.etlmaster.core.DataSourceHolder;
import com.etlmaster.executor.utils.LogWriter;
import com.etlmaster.jdbc.JdbcTemplate;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public class HiveJdbcUtil {
	private static final String ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME = "Client";
	private static final String ZOOKEEPER_SERVER_PRINCIPAL_KEY = "zookeeper.server.principal";
	private static final String ZOOKEEPER_DEFAULT_SERVER_PRINCIPAL = "zookeeper/hadoop";

	private static Configuration CONF = null;
	private static String KRB5_FILE_PARAM = "krb5file";
	private static String USER_NAME_PARAM = "username";
	private static String USER_KEYTAB_FILE_PARAME = "keytabfile";
	private static String ZK_LOGIN_NAME_PARAM = "zkloginname";
	private static String ZK_SERVER_PRINCIPAL_KEY_PARAM = "zkprincipalkey";
	private static String ZK_PRINCIPAL_PARAM = "zkprincipal";

	private static String KRB5_FILE = null;
	private static String USER_NAME = null;
	private static String USER_KEYTAB_FILE = null;
	private static String ZK_LOGIN_NAME = null;
	private static String ZK_SERVER_PRINCIPAL_KEY = null;
	private static String ZK_PRINCIPAL = null;

	private static void init() throws IOException {
		CONF = new Configuration();
		USER_NAME = System.getProperty(USER_NAME_PARAM);
		USER_KEYTAB_FILE = System.getProperty(USER_KEYTAB_FILE_PARAME);
		KRB5_FILE = System.getProperty(KRB5_FILE_PARAM);
		ZK_LOGIN_NAME = System.getProperty(ZK_LOGIN_NAME_PARAM,
				ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME);
		ZK_SERVER_PRINCIPAL_KEY = System.getProperty(
				ZK_SERVER_PRINCIPAL_KEY_PARAM, ZOOKEEPER_SERVER_PRINCIPAL_KEY);
		ZK_PRINCIPAL = System.getProperty(ZK_PRINCIPAL_PARAM,
				ZOOKEEPER_DEFAULT_SERVER_PRINCIPAL);
		LogWriter.addLog("INFO", "认证配置文件：[{}],认证文件：[{}]", KRB5_FILE,
				USER_KEYTAB_FILE);
		LogWriter.addLog("DEBUG", "认证zk登录名：[{}],zk认证键：[{}],zk认证值：[{}]",
				ZK_LOGIN_NAME, ZK_SERVER_PRINCIPAL_KEY, ZK_PRINCIPAL);
		LoginUtil.setJaasConf(ZK_LOGIN_NAME, USER_NAME, USER_KEYTAB_FILE);
		LogWriter.addLog("DEBUG", "设置JAAS配置完成...");
		LoginUtil.setZookeeperServerPrincipal(ZK_SERVER_PRINCIPAL_KEY,
				ZK_PRINCIPAL);
		LogWriter.addLog("DEBUG", "设置zk认证配置完成...");
	}

	// 链接华为
	public static Connection getConn(String driverClassName, String url,
			String krbconf, String userName, String passWord)
			throws ClassNotFoundException, SQLException {

		if (krbconf != null&&krbconf.trim().length()>0) {
			url = ImpalaUtil.dealUrl(url);
			try {
				init();
				LogWriter.addLog("DEBUG", "认证初始化完成...");
				LoginUtil.login(USER_NAME, USER_KEYTAB_FILE,
						null != KRB5_FILE ? KRB5_FILE : krbconf, CONF);
				LogWriter.addLog("DEBUG", "认证登录成功...");
			} catch (IOException e) {
				LogWriter.addLog("ERROR", "认证登录失败：", e.getMessage());
			}
		}
		// 加载Hive JDBC驱动
		Class.forName(driverClassName);
		Connection connection = null;
		connection = DriverManager.getConnection(url, userName, passWord);

		return connection;
	}

	public static Connection getConnection(String driverName,
			String connectUrl, String userName, String password)
			throws Exception {
		Connection conn = null;
		Class.forName(driverName);
		conn = DriverManager.getConnection(connectUrl, userName, password);
		return conn;
	}

	public static boolean query(String sql, String dsName) {

		boolean flag = true;
		Connection conn = null;
		Statement stmt = null;
		try {
			DataSourceDefine dfs = DataSourceHolder.getInstance()
					.getDataSourceDefine(dsName);
			if (dfs == null) {
				LogWriter.addLog("ERROR",
						"执行hiveSql取团队数据源为空，请检查程序归属与数据库资源是否配置一致");
				return false;
			}
			// 增加impala连接
			if (ImpalaUtil.isImpala(dfs.getUrl())) {
				conn = ImpalaUtil.getConnection(dfs.getDriverClassName(),
						dfs.getUrl(), dfs.getUsername(), dfs.getPassword());
			} else if (null != dfs.getKrbconf()
					&& dfs.getKrbconf().trim().length() > 2) {// 如果是华为kerberos认证则需要使用特殊的链接方法
				conn = getConn(dfs.getDriverClassName(), dfs.getUrl(), dfs
						.getKrbconf().trim(), dfs.getUsername(),
						dfs.getPassword());
			} else {
				JdbcTemplate jdbcTemplate = new JdbcTemplate(dsName);
				conn = jdbcTemplate.getDataSource().getConnection();
			}
			stmt = conn.createStatement();
			stmt.execute(sql);
		} catch (Exception e) {
			LogWriter.addLog("ERROR", e.getMessage());
			flag = false;
		} finally {
			try {
				if (stmt != conn)
					conn.close();
				if (null != conn)
					conn.close();
			} catch (Exception e1) {
			}
		}
		return flag;
	}

	public static String execSql(String[] sqls, String dsName) {
		String flag = "success";
		Connection conn = null;
		Statement stmt = null;
		try {
			DataSourceDefine dfs = DataSourceHolder.getInstance()
					.getDataSourceDefine(dsName);
			if (dfs == null) {
				LogWriter.addLog("ERROR",
						"执行hiveSql取团队数据源为空，请检查程序归属与数据库资源是否配置一致");
				return "执行hiveSql取团队数据源为空，请检查程序归属与数据库资源是否配置一致";
			}

			// 增加impala连接
			if (ImpalaUtil.isImpala(dfs.getUrl())) {
				conn = ImpalaUtil.getConnection(dfs.getDriverClassName(),
						dfs.getUrl(), dfs.getUsername(), dfs.getPassword());
			} else if (null != dfs.getKrbconf()
					&& dfs.getKrbconf().trim().length() > 2) {// 如果是华为kerberos认证则需要使用特殊的链接方法
				conn = getConn(dfs.getDriverClassName(), dfs.getUrl(), dfs
						.getKrbconf().trim(), dfs.getUsername(),
						dfs.getPassword());
			} else {
				JdbcTemplate jdbcTemplate = new JdbcTemplate(dsName);
				conn = jdbcTemplate.getDataSource().getConnection();
			}
			stmt = conn.createStatement();
			for (String sSql : sqls) {
				if (sSql.length() > 0) {
					LogWriter.addLog("INFO",
							"在 数据库:{}执行sql:\n               {}", dsName, sSql);
					stmt.execute(sSql);
				}
			}
		} catch (Exception e) {
			LogWriter.addLog("ERROR", e.getMessage());
			flag = e.getMessage();
		} finally {
			try {
				if (stmt != conn)
					conn.close();
				if (null != conn)
					conn.close();
			} catch (Exception e1) {
			}
		}
		return flag;
	}

	public static List queryForList(String sql, String dsName)
			throws SQLException, ClassNotFoundException, LoginException,
			PrivilegedActionException {
		DataSourceDefine dfs = DataSourceHolder.getInstance()
				.getDataSourceDefine(dsName);
		Connection conn = null;
		// 增加impala连接
		if (ImpalaUtil.isImpala(dfs.getUrl())) {
			conn = ImpalaUtil.getConnection(dfs.getDriverClassName(),
					dfs.getUrl(), dfs.getUsername(), dfs.getPassword());
		} else if (null != dfs.getKrbconf()
				&& dfs.getKrbconf().trim().length() > 2) {// 如果是华为kerberos认证则需要使用特殊的链接方法
			conn = getConn(dfs.getDriverClassName(), dfs.getUrl(), dfs
					.getKrbconf().trim(), dfs.getUsername(), dfs.getPassword());
		} else {
			JdbcTemplate jdbcTemplate = new JdbcTemplate(dsName);
			conn = jdbcTemplate.getDataSource().getConnection();
		}
		Statement stmt = conn.createStatement();
		return resultSetToList(stmt.executeQuery(sql));
	}

	private static List resultSetToList(ResultSet rs)
			throws java.sql.SQLException {
		if (rs == null)
			return Collections.EMPTY_LIST;
		ResultSetMetaData md = rs.getMetaData();
		int columnCount = md.getColumnCount();
		List list = new ArrayList();
		Map rowData = new HashMap();
		while (rs.next()) {
			rowData = new HashMap(columnCount);
			for (int i = 1; i <= columnCount; i++) {
				String columnname = md.getColumnName(i);
				int actualNameIndex = columnname.indexOf(".") + 1;
				columnname = columnname.substring(actualNameIndex);
				rowData.put(columnname, rs.getObject(i));
			}
			list.add(rowData);
		}
		return list;
	}
}
